package test;

import java.io.*;
import java.util.ArrayList;
import java.util.Date;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.lsda.job.ActionHandler;
import org.lsda.util.*;


public class JobExecution implements MessageListener, ExceptionListener {

	private Log log = LogFactory.getLog(JobExecution.class);

	private String queueName = Util.getBRQueName();

	private String url = Util.getBRUrl();

	private String userName = Util.getBRUser();

	private String password = Util.getBRPasswd();

	private Session session = null;

	private MessageProducer replyProducer = null;

	private ArrayList<ActionHandler> actionHandlers = new ArrayList<ActionHandler>();

	public JobExecution() {
	}

	public String getQueueName() {
		return queueName;
	}

	public void setQueueName(String queueName) {
		this.queueName = queueName;
	}

	public String getUrl() {
		return url;
	}

	public void setUrl(String url) {
		this.url = url;
	}

	public String getUserName() {
		return userName;
	}

	public void setUserName(String userName) {
		this.userName = userName;
	}

	public String getPassword() {
		return password;
	}

	public void setPassword(String password) {
		this.password = password;
	}

	public void addActionHandler(ActionHandler ah) {
		actionHandlers.add(ah);
	}

	public void clearActionHandler() {
		actionHandlers.clear();
	}

	private void executeJob() {
		Connection connection = null;
		try {
			ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
					userName, password, url);
			connection = connectionFactory.createConnection();
			connection.setExceptionListener(this);
			connection.start();

			session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);

			Destination destination = session.createQueue(queueName);

			MessageConsumer consumer = session.createConsumer(destination);
			consumer.setMessageListener(this);
		} catch (Exception ex) {
			ex.printStackTrace();
		}
	}

	public boolean registerAction(Job job) {
		boolean ret = false;
		String actsql = "insert into action(jobid,actid,prgname,param,status,begtime)values(?,?,?,?,-1,now())";
		String jobsql = "update job set rsptime=now() where jobid=?";
		java.sql.Connection conn = null;
		try {
			conn = Util.getConnection();
			conn.setAutoCommit(false);
			java.sql.PreparedStatement pstmt = conn.prepareStatement(actsql);

			// register all actions of job
			Action act = null;
			for (int i = 0; i < job.getSize(); i++) {
				act = job.getAction(i);
				pstmt.setInt(1, act.getJobId());
				pstmt.setInt(2, act.getSubJobId());
				pstmt.setString(3, act.getPrgName());
				pstmt.setString(4, act.getParam());
				pstmt.execute();
			}

			// update job response time
			java.sql.PreparedStatement jpstmt = conn.prepareStatement(jobsql);
			jpstmt.setInt(1, job.getJobId());
			jpstmt.execute();

			conn.commit();
			pstmt.close();
			jpstmt.close();
			ret = true;
		} catch (Exception ex) {
			log.error(ex.getMessage());
		} finally {
			try {
				conn.close();
			} catch (Throwable igrone) {
			}
		}
		return ret;
	}

	public boolean updateActionStatus(Action act) {
		boolean ret = false;
		String jobsql = "update action set status=?,info=?,endtime=now() where jobid=? and actid=?";
		java.sql.Connection conn = null;
		try {
			conn = Util.getConnection();
			conn.setAutoCommit(false);
			java.sql.PreparedStatement pstmt = conn.prepareStatement(jobsql);
			pstmt.setInt(1, act.getStatus());
			pstmt.setString(2, act.getInfo());
			pstmt.setInt(3, act.getJobId());
			pstmt.setInt(4, act.getSubJobId());
			pstmt.execute();
			conn.commit();
			pstmt.close();
			ret = true;
		} catch (Exception ex) {
			log.error(ex.getMessage());
		} finally {
			try {
			} catch (Throwable ignore) {
			}
		}
		return ret;
	}

	@Override
	public void onException(JMSException arg0) {
		// TODO Auto-generated method stub
		log.error(arg0.getMessage());
		System.exit(-1);
	}

	public Action doAct(Action act) {
		String cmd = act.getPrgName() + " " + act.getParam();
		log.info(cmd);
		StringBuffer sbinfo = new StringBuffer();
		String line =null;
		int status = -1;
		try {
			Runtime rt = Runtime.getRuntime();
			Process proc = rt.exec(cmd);
			BufferedReader berr = new BufferedReader(new InputStreamReader(
					proc.getErrorStream()));
			BufferedReader bout = new BufferedReader(new InputStreamReader(
					proc.getInputStream()));
			while((line=berr.readLine())!=null){
				sbinfo.append("stderr:");
				sbinfo.append(line);
				sbinfo.append("\n");
			}
			
			berr.close();
			
			while((line=bout.readLine())!=null){
				sbinfo.append("stdout:");
				sbinfo.append(line);
				sbinfo.append("\n");
			}
			
			bout.close();
			
			status = proc.waitFor();
			//make sure that the status should be non-negative
			status = Math.abs(status);
			
		} catch (Exception ex) {
			sbinfo.append("stderr:"+ex.getMessage());
			log.error(ex.getMessage());
			status = 2;
		}
		act.setStatus(status);
		act.setEndTime(new Date());
		act.setInfo(sbinfo.toString());
		log.info(act.getInfo());
		return act;
	}

	@Override
	public void onMessage(Message arg0) {
		// TODO Auto-generated method stub
		if (arg0 instanceof ObjectMessage) {
			try {
				ObjectMessage om = (ObjectMessage) arg0;

				Job job = (Job) om.getObject();
				registerAction(job);
				replyProducer = session.createProducer(om.getJMSReplyTo());
				for (int i = 0; i < job.getSize(); i++) {
					Action act = job.getAction(i);
					act.setStatus(1);
					act.setInfo("running");
					updateActionStatus(act);
					Action act1 = doAct(act);
					log.info("check 1:"+act1.getInfo());
					updateActionStatus(act);
					ObjectMessage objMsg = session.createObjectMessage(act);
					objMsg.setJMSCorrelationID("" + job.getJobId());
					replyProducer.send(objMsg);
					session.commit();
					if (act.getStatus() != 0) {
						break;
					}
				}
				replyProducer.close();
			} catch (JMSException e) {
				// TODO Auto-generated catch block
				log.error(e.getMessage());
			}
		}
	}

	public static void main(String args[]) {
		JobExecution je = new JobExecution();
		je.executeJob();
	}

}
